package com.example.yckjbigdataflink.utils;

import org.apache.flink.api.common.RuntimeExecutionMode;
/*import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

*//**
 * 依赖
 *
 *     <!-- 1. Flink核心流处理依赖 -->
 *     <dependency>
 *         <groupId>org.apache.flink</groupId>
 *         <artifactId>flink-streaming-java</artifactId>
 *         <version>1.17.1</version>
 *     </dependency>
 *
 *     <!-- 2. RocksDB状态后端依赖 (必需) -->
 *     <dependency>
 *         <groupId>org.apache.flink</groupId>
 *         <artifactId>flink-statebackend-rocksdb</artifactId>
 *         <version>1.17.1</version>
 *     </dependency>
 *
 *     <!-- 3. 如果使用HDFS存储检查点 (可选) -->
 *     <dependency>
 *         <groupId>org.apache.flink</groupId>
 *         <artifactId>flink-hadoop-fs</artifactId>
 *         <version>1.17.1</version>
 *     </dependency>
 *
 *     <!-- 4. 如果需要S3存储 (可选) -->
 *     <dependency>
 *         <groupId>org.apache.flink</groupId>
 *         <artifactId>flink-s3-fs-hadoop</artifactId>
 *         <version>1.17.1</version>
 *     </dependency>
 *
 *
 *
 * Flink 1.13+ 检查点配置完整示例
 * 展示新版本中正确的状态后端和检查点配置方法
 *//*
public class FlinkCheckpointConfigExample {

    *//**
     * 方法1：使用HashMapStateBackend + 文件系统检查点存储 (推荐用于小状态)
     *//*
    public static StreamExecutionEnvironment setupHashMapStateBackend() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1. 设置状态后端为HashMapStateBackend（替代MemoryStateBackend）
        env.setStateBackend(new HashMapStateBackend());

        // 2. 设置检查点存储位置
        env.getCheckpointConfig().setCheckpointStorage("hdfs:///flink/checkpoints");

        // 3. 启用检查点，设置检查点间隔
        env.enableCheckpointing(60000); // 60秒

        // 4. 配置检查点详细参数
        configureCheckpointDetails(env);

        return env;
    }

    *//**
     * 方法2：使用EmbeddedRocksDBStateBackend (推荐用于大状态)
     *//*
    public static StreamExecutionEnvironment setupRocksDBStateBackend() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1. 设置状态后端为RocksDB
        EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true);

        // RocksDB优化配置
        rocksDBStateBackend.setPredefinedOptions(
                org.apache.flink.contrib.streaming.state.PredefinedOptions.SPINNING_DISK_OPTIMIZED
        );

        env.setStateBackend(rocksDBStateBackend);

        // 2. 设置检查点存储位置
        env.getCheckpointConfig().setCheckpointStorage("hdfs:///flink/checkpoints");

        // 3. 启用检查点
        env.enableCheckpointing(30000); // 30秒，RocksDB可以更频繁检查点

        // 4. 配置检查点详细参数
        configureCheckpointDetails(env);

        return env;
    }

    *//**
     * 方法3：通过配置文件设置（生产环境推荐）
     *//*
    public static StreamExecutionEnvironment setupFromConfiguration() {
        // 创建配置对象
        Configuration config = new Configuration();

        // 状态后端配置
        config.setString("state.backend", "hashmap"); // 或 "rocksdb"
        config.setString("state.checkpoints.dir", "hdfs:///flink/checkpoints");
        config.setString("state.savepoints.dir", "hdfs:///flink/savepoints");

        // 检查点配置
        config.set(org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTING_INTERVAL,
                Duration.ofSeconds(60));
        config.set(org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTING_MODE,
                CheckpointingMode.EXACTLY_ONCE);
        config.set(org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTING_TIMEOUT,
                Duration.ofMinutes(10));
        config.set(org.apache.flink.configuration.CheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS, 1);
        config.set(org.apache.flink.configuration.CheckpointingOptions.MIN_PAUSE_BETWEEN_CHECKPOINTS,
                Duration.ofSeconds(500));

        // RocksDB特定配置（如果使用RocksDB）
        config.setString("state.backend.rocksdb.predefined-options", "SPINNING_DISK_OPTIMIZED");
        config.set(org.apache.flink.configuration.RocksDBOptions.USE_MANAGED_MEMORY, true);

        // 使用配置创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);

        return env;
    }

    *//**
     * 通用检查点详细配置
     *//*
    private static void configureCheckpointDetails(StreamExecutionEnvironment env) {
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        // 检查点模式：精确一次
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // 检查点超时时间
        checkpointConfig.setCheckpointTimeout(600000); // 10分钟

        // 最大并发检查点数量
        checkpointConfig.setMaxConcurrentCheckpoints(1);

        // 两次检查点之间的最小暂停时间
        checkpointConfig.setMinPauseBetweenCheckpoints(500);

        // 检查点失败时是否让任务失败
        checkpointConfig.setTolerableCheckpointFailureNumber(3);

        // 作业取消时保留检查点
        checkpointConfig.setExternalizedCheckpointCleanup(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // 启用检查点压缩（可选）
        checkpointConfig.setCheckpointCompression(true);

        // 配置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(
                3, // 最大失败次数
                Time.minutes(5), // 时间窗口
                Time.seconds(30) // 重启间隔
        ));
    }

    *//**
     * 高可用和性能优化配置
     *//*
    public static StreamExecutionEnvironment setupHighAvailabilityConfig() {
        Configuration config = new Configuration();

        // === 状态后端配置 ===
        config.setString("state.backend", "rocksdb");
        config.setString("state.checkpoints.dir", "hdfs:///flink/checkpoints");
        config.setString("state.savepoints.dir", "hdfs:///flink/savepoints");

        // === 检查点配置 ===
        config.set(org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTING_INTERVAL,
                Duration.ofSeconds(30));
        config.set(org.apache.flink.configuration.CheckpointingOptions.CHECKPOINTING_TIMEOUT,
                Duration.ofMinutes(15));
        config.set(org.apache.flink.configuration.CheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS, 1);
        config.set(org.apache.flink.configuration.CheckpointingOptions.MIN_PAUSE_BETWEEN_CHECKPOINTS,
                Duration.ofSeconds(500));
        config.set(org.apache.flink.configuration.CheckpointingOptions.TOLERABLE_FAILURE_NUMBER, 5);

        // === RocksDB性能优化 ===
        config.setString("state.backend.rocksdb.predefined-options", "SPINNING_DISK_OPTIMIZED");
        config.set(org.apache.flink.configuration.RocksDBOptions.USE_MANAGED_MEMORY, true);

        // 内存配置
        config.set(org.apache.flink.configuration.TaskManagerOptions.MANAGED_MEMORY_SIZE,
                MemorySize.parse("1gb"));
        config.setDouble("state.backend.rocksdb.memory.managed.fraction", 0.6);

        // === 高可用配置 ===
        config.setString("high-availability", "zookeeper");
        config.setString("high-availability.zookeeper.quorum", "zk1:2181,zk2:2181,zk3:2181");
        config.setString("high-availability.storageDir", "hdfs:///flink/ha");
        config.setString("high-availability.zookeeper.path.root", "/flink");

        // === 网络配置 ===
        config.setString("taskmanager.network.memory.fraction", "0.1");
        config.setString("taskmanager.network.memory.min", "64mb");
        config.setString("taskmanager.network.memory.max", "1gb");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);

        return env;
    }

    *//**
     * 本地测试环境配置
     *//*
    public static StreamExecutionEnvironment setupLocalTestEnvironment() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 本地测试使用内存状态后端
        env.setStateBackend(new HashMapStateBackend());

        // 本地文件系统检查点
        env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");

        // 较短的检查点间隔用于测试
        env.enableCheckpointing(5000); // 5秒

        // 简化的检查点配置
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        checkpointConfig.setCheckpointTimeout(60000); // 1分钟
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        checkpointConfig.setExternalizedCheckpointCleanup(
                CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

        return env;
    }

    *//**
     * 示例主方法
     *//*
    public static void main(String[] args) throws Exception {
        // 根据不同场景选择合适的配置
        StreamExecutionEnvironment env;

        String profile = System.getProperty("profile", "local");

        switch (profile) {
            case "prod":
                env = setupHighAvailabilityConfig();
                break;
            case "test":
                env = setupRocksDBStateBackend();
                break;
            case "local":
            default:
                env = setupLocalTestEnvironment();
                break;
        }

        // 设置并行度
        env.setParallelism(4);

        // 添加你的数据流处理逻辑
        env.fromElements("Hello", "World", "Flink")
                .map(String::toUpperCase)
                .print();

        // 执行作业
        env.execute("Flink Checkpoint Example");
    }
}

*//**
 * 检查点监控和管理工具类
 *//*
class CheckpointMonitor {

    *//**
     * 获取检查点统计信息
     *//*
    public static void printCheckpointStats(StreamExecutionEnvironment env) {
        CheckpointConfig config = env.getCheckpointConfig();

        System.out.println("=== 检查点配置信息 ===");
        System.out.println("检查点间隔: " + config.getCheckpointInterval() + "ms");
        System.out.println("检查点超时: " + config.getCheckpointTimeout() + "ms");
        System.out.println("检查点模式: " + config.getCheckpointingMode());
        System.out.println("最大并发检查点: " + config.getMaxConcurrentCheckpoints());
        System.out.println("检查点失败容忍次数: " + config.getTolerableCheckpointFailureNumber());
    }

    *//**
     * 检查点健康检查
     *//*
    public static boolean isCheckpointHealthy(long lastCheckpointTime, long checkpointInterval) {
        long currentTime = System.currentTimeMillis();
        long timeSinceLastCheckpoint = currentTime - lastCheckpointTime;

        // 如果超过检查点间隔的3倍还没有新检查点，认为不健康
        return timeSinceLastCheckpoint < (checkpointInterval * 3);
    }
}*/
